package com.raven.play.stream.test.javadsl.tutorial;

import akka.Done;
import akka.NotUsed;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.Behaviors;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Keep;

import java.util.Arrays;
import java.util.concurrent.CompletionStage;

public class AkkaStreamsExample {
    public static void main(String[] args) {
        ActorSystem<Void> actorSystem = ActorSystem.create(Behaviors.empty(), "actorSystem");

        Source<Integer, NotUsed> source = Source.from(Arrays.asList(1, 2, 3, 4, 5));
        Flow<Integer, Integer, NotUsed> flow = Flow.of(Integer.class).map(n -> n * 2);
        Sink<Integer, CompletionStage<Done>> sink = Sink.foreach(System.out::println);

        CompletionStage<Done> doneStage = source.via(flow).toMat(sink, Keep.right()).run(actorSystem);
        doneStage.whenComplete((done, ex) -> {
            if (ex != null) {
                System.out.println("Stream finished with error: " + ex.getMessage());
            } else {
                System.out.println("Stream finished successfully.");
            }
            actorSystem.terminate();
            if (true) {
                throw new RuntimeException("hahah");
            }

        });
    }
}
